/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2022/07/05.
//

#include "sql/expr/expression.h"
#include "sql/expr/tuple.h"
#include "sql/expr/arithmetic_operator.hpp"

using namespace std;

#include <string>
using namespace std;

bool is_like(const string& a, const string& b) {
    int a_len = a.length();
    int b_len = b.length();
    int a_idx = 0, b_idx = 0;
    int a_back = -1, b_back = -1; // 用于回溯的位置
    
    while (a_idx < a_len) {
        if (b_idx < b_len) {
            char b_char = b[b_idx];
            
            if (b_char == '%') {
                // 记录回溯点
                a_back = a_idx;
                b_back = b_idx;
                b_idx++;
                continue;
            } else if (b_char == '_') {
                // 匹配一个非单引号字符
                if (a[a_idx] != '\'') {
                    a_idx++;
                    b_idx++;
                    continue;
                }
            } else {
                // 精确匹配
                if (a[a_idx] == b_char) {
                    a_idx++;
                    b_idx++;
                    continue;
                }
            }
        }
        
        // 不匹配时尝试回溯
        if (b_back != -1) {
            a_idx = a_back + 1;
            b_idx = b_back;
            a_back = a_idx;
            continue;
        }
        
        return false;
    }
    
    // 处理b中剩余的%
    while (b_idx < b_len && b[b_idx] == '%') {
        b_idx++;
    }
    
    return b_idx == b_len;
}



RC FieldExpr::get_value(const Tuple &tuple, Value &value) const
{
  return tuple.find_cell(TupleCellSpec(table_name(), field_name()), value);
}

bool FieldExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (other.type() != ExprType::FIELD) {
    return false;
  }
  const auto &other_field_expr = static_cast<const FieldExpr &>(other);
  return table_name() == other_field_expr.table_name() && field_name() == other_field_expr.field_name();
}

// TODO: 在进行表达式计算时，`chunk` 包含了所有列，因此可以通过 `field_id` 获取到对应列。
// 后续可以优化成在 `FieldExpr` 中存储 `chunk` 中某列的位置信息。
RC FieldExpr::get_column(Chunk &chunk, Column &column)
{
  if (pos_ != -1) {
    column.reference(chunk.column(pos_));
  } else {
    column.reference(chunk.column(field().meta()->field_id()));
  }
  return RC::SUCCESS;
}

bool ValueExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (other.type() != ExprType::VALUE) {
    return false;
  }
  const auto &other_value_expr = static_cast<const ValueExpr &>(other);
  return value_.compare(other_value_expr.get_value()) == 0;
}

RC ValueExpr::get_value(const Tuple &tuple, Value &value) const
{
  value = value_;
  return RC::SUCCESS;
}

RC ValueExpr::get_column(Chunk &chunk, Column &column)
{
  column.init(value_);
  return RC::SUCCESS;
}

/////////////////////////////////////////////////////////////////////////////////
CastExpr::CastExpr(unique_ptr<Expression> child, AttrType cast_type) : child_(std::move(child)), cast_type_(cast_type)
{}

CastExpr::~CastExpr() {}

RC CastExpr::cast(const Value &value, Value &cast_value) const
{
  RC rc = RC::SUCCESS;
  if (this->value_type() == value.attr_type()) {
    cast_value = value;
    return rc;
  }
  rc = Value::cast_to(value, cast_type_, cast_value);
  return rc;
}

RC CastExpr::get_value(const Tuple &tuple, Value &result) const
{
  Value value;
  RC rc = child_->get_value(tuple, value);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  return cast(value, result);
}

RC CastExpr::try_get_value(Value &result) const
{
  Value value;
  RC rc = child_->try_get_value(value);
  if (rc != RC::SUCCESS) {
    return rc;
  }

  return cast(value, result);
}

////////////////////////////////////////////////////////////////////////////////

ComparisonExpr::ComparisonExpr(CompOp comp, unique_ptr<Expression> left, unique_ptr<Expression> right)
    : comp_(comp), left_(std::move(left)), right_(std::move(right))
{
  // if (left_->type()==ExprType::UNBOUND_FIELD)
  // {
  //   left_ = std::move((std::move(left_)));
  // }
}

ComparisonExpr::~ComparisonExpr() {}

RC ComparisonExpr::compare_value(const Value &left, const Value &right, bool &result) const
{
  RC  rc         = RC::SUCCESS;
  if(comp_ == CompOp::LIKE_OP || comp_ == CompOp::NOT_LIKE_OP) {
    // Implement LIKE comparison logic here
    if(left.attr_type() != AttrType::CHARS || right.attr_type() != AttrType::CHARS) {
      LOG_WARN("LIKE operator only supports CHARS type");
      return RC::INVALID_ARGUMENT;
    }
    // For simplicity, we assume that the right side is a pattern and the left side is a string
    result = is_like(left.get_string(), right.get_string());
    if (comp_ == CompOp::NOT_LIKE_OP) {
      result = !result; // NOT LIKE
    }
    return rc;
  }
  //------------------------------ NULL Comparison ---------------------------------------
  if(left.attr_type() == AttrType::NULLS || right.attr_type() == AttrType::NULLS) {
    if (comp_ == CompOp::IS_OP) 
    {
      result = (left.attr_type() == AttrType::NULLS && right.attr_type() == AttrType::NULLS);
    } 
    else if (comp_ == CompOp::IS_NOT_OP) 
    {
      result = !(left.attr_type() == AttrType::NULLS && right.attr_type() == AttrType::NULLS);
    } 
    else 
    {
      result = false;
    }
    return rc;
  }
  //-------------------------------------------------------------------------------------
  int cmp_result = left.compare(right);
  result         = false;
  switch (comp_) {
    case EQUAL_TO: {
      result = (0 == cmp_result);
    } break;
    case LESS_EQUAL: {
      result = (cmp_result <= 0);
    } break;
    case NOT_EQUAL: {
      result = (cmp_result != 0);
    } break;
    case LESS_THAN: {
      result = (cmp_result < 0);
    } break;
    case GREAT_EQUAL: {
      result = (cmp_result >= 0);
    } break;
    case GREAT_THAN: {
      result = (cmp_result > 0);
    } break;
    default: {
      LOG_WARN("unsupported comparison. %d", comp_);
      rc = RC::INTERNAL;
    } break;
  }

  return rc;
}

RC ComparisonExpr::try_get_value(Value &cell) const
{
  if (left_->type() == ExprType::VALUE && right_->type() == ExprType::VALUE) {
    ValueExpr *  left_value_expr  = static_cast<ValueExpr *>(left_.get());
    ValueExpr *  right_value_expr = static_cast<ValueExpr *>(right_.get());
    const Value &left_cell        = left_value_expr->get_value();
    const Value &right_cell       = right_value_expr->get_value();

    bool value = false;
    RC   rc    = compare_value(left_cell, right_cell, value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to compare tuple cells. rc=%s", strrc(rc));
    } else {
      cell.set_boolean(value);
    }
    return rc;
  }

  return RC::INVALID_ARGUMENT;
}

RC ComparisonExpr::get_value(const Tuple &tuple, Value &value) const
{
  
  Value left_value;
  Value right_value;
  RC rc = RC::SUCCESS;
  if(left_ == nullptr) {

    ASSERT(comp_==CompOp::EXISTS_OP || comp_==CompOp::NOT_EXISTS_OP,"no left while comp is not Exists");
    rc = right_->get_value(tuple, right_value);
    if (rc == RC::MULTI_COLUMN_SUBQUERY || rc== RC::SUCCESS)
    {
      value.set_boolean(comp_ == CompOp::EXISTS_OP);
    }
    static_cast<SubqueryExpr *>(right_.get())->close();
    return rc;
  }
  else if(left_->type()==ExprType::SUBQUERY ) {//&& right_->type()==ExprType::SUBQUERY) {
    rc = left_->get_value(tuple, left_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
      static_cast<SubqueryExpr *>(left_.get())->close();
      return rc;
    }
    rc = left_->get_value(tuple,left_value);
    if (rc != RC::RECORD_EOF)
    {
      LOG_WARN("left subquery must return one row");
      static_cast<SubqueryExpr *>(left_.get())->close();
      return RC::SUBQUERY_INVALID_ROW_NUM;
    }
  }
  else if(left_->type() == ExprType::VALUE_LIST) {
    return RC::UNSUPPORTED;
  }
  else 
  {
    rc = left_->get_value(tuple, left_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
      return rc;
    }
  }
  ASSERT(comp_!=CompOp::EXISTS_OP && comp_!=CompOp::NOT_EXISTS_OP,"comp is Exists but left is not nullptr");

  if(comp_==CompOp::IN_OP || comp_==CompOp::NOT_IN_OP) {
    ASSERT(right_->type()==ExprType::SUBQUERY || right_->type()==ExprType::VALUE_LIST,
             "IN_OP or NOT_IN_OP must have subquery or value_list on right");
    if(left_value.attr_type()==AttrType::NULLS) {
      value.set_boolean(false);
      return rc;
    }
    while(rc==RC::SUCCESS)
    {
        rc = right_->get_value(tuple, right_value);
        if(rc == RC::RECORD_EOF) {
            rc = RC::SUCCESS;
            break;
        }
        else if(rc != RC::SUCCESS)
        {
            LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
            return rc;
        }
        int cmp = left_value.compare(right_value);
        if(cmp==0)
        {
            value.set_boolean(CompOp::IN_OP == comp_);
            if(right_->type() == ExprType::SUBQUERY)
              static_cast<SubqueryExpr *>(right_.get())->close();
            else if(right_->type() == ExprType::VALUE_LIST)
              static_cast<ValueListExpr *>(right_.get())->close();
            return rc;
        }
        
    }
    value.set_boolean(comp_ == CompOp::NOT_IN_OP);
    return rc;
      
  }
  ASSERT(right_->type() != ExprType::VALUE_LIST, "right expression must not be VALUE_LIST for comparison");


  if(right_->type() == ExprType::SUBQUERY) {
    rc = right_->get_value(tuple, right_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      static_cast<SubqueryExpr *>(right_.get())->close();
      return rc;
    }
    rc = right_->get_value(tuple, right_value);
    if(rc!=RC::RECORD_EOF) {
      LOG_WARN("right subquery must return one row");
      static_cast<SubqueryExpr *>(right_.get())->close();
      return RC::SUBQUERY_INVALID_ROW_NUM;
    }
  }
  else 
  {
    rc = right_->get_value(tuple, right_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
  }


  // rc = left_->get_value(tuple, left_value);
  // if (rc != RC::SUCCESS) {
  //   LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
  //   return rc;
  // }
  // rc = right_->get_value(tuple, right_value);
  // if (rc != RC::SUCCESS) {
  //   LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
  //   return rc;
  // }

  bool bool_value = false;

  rc = compare_value(left_value, right_value, bool_value);
  if (rc == RC::SUCCESS) {
    value.set_boolean(bool_value);
  }
  return rc;
}

RC ComparisonExpr::eval(Chunk &chunk, vector<uint8_t> &select)
{
  RC     rc = RC::SUCCESS;
  Column left_column;
  Column right_column;

  rc = left_->get_column(chunk, left_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_column(chunk, right_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
    return rc;
  }
  if (left_column.attr_type() != right_column.attr_type()) {
    LOG_WARN("cannot compare columns with different types");
    return RC::INTERNAL;
  }
  if (left_column.attr_type() == AttrType::INTS) {
    rc = compare_column<int>(left_column, right_column, select);
  } else if (left_column.attr_type() == AttrType::FLOATS) {
    rc = compare_column<float>(left_column, right_column, select);
  } else {
    // TODO: support string compare
    LOG_WARN("unsupported data type %d", left_column.attr_type());
    return RC::INTERNAL;
  }
  return rc;
}

template <typename T>
RC ComparisonExpr::compare_column(const Column &left, const Column &right, vector<uint8_t> &result) const
{
  RC rc = RC::SUCCESS;

  bool left_const  = left.column_type() == Column::Type::CONSTANT_COLUMN;
  bool right_const = right.column_type() == Column::Type::CONSTANT_COLUMN;
  if (left_const && right_const) {
    compare_result<T, true, true>((T *)left.data(), (T *)right.data(), left.count(), result, comp_);
  } else if (left_const && !right_const) {
    compare_result<T, true, false>((T *)left.data(), (T *)right.data(), right.count(), result, comp_);
  } else if (!left_const && right_const) {
    compare_result<T, false, true>((T *)left.data(), (T *)right.data(), left.count(), result, comp_);
  } else {
    compare_result<T, false, false>((T *)left.data(), (T *)right.data(), left.count(), result, comp_);
  }
  return rc;
}

////////////////////////////////////////////////////////////////////////////////
ConjunctionExpr::ConjunctionExpr(Type type, vector<unique_ptr<Expression>> &children)
    : conjunction_type_(type), children_(std::move(children))
{}

RC ConjunctionExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;
  if (children_.empty()) {
    value.set_boolean(true);
    return rc;
  }

  Value tmp_value;
  for (const unique_ptr<Expression> &expr : children_) {
    rc = expr->get_value(tuple, tmp_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value by child expression. rc=%s", strrc(rc));
      return rc;
    }
    bool bool_value = tmp_value.get_boolean();
    if ((conjunction_type_ == Type::AND && !bool_value) || (conjunction_type_ == Type::OR && bool_value)) {
      value.set_boolean(bool_value);
      return rc;
    }
  }

  bool default_value = (conjunction_type_ == Type::AND);
  value.set_boolean(default_value);
  return rc;
}

////////////////////////////////////////////////////////////////////////////////

ArithmeticExpr::ArithmeticExpr(ArithmeticExpr::Type type, Expression *left, Expression *right)
    : arithmetic_type_(type), left_(left), right_(right)
{}
ArithmeticExpr::ArithmeticExpr(ArithmeticExpr::Type type, unique_ptr<Expression> left, unique_ptr<Expression> right)
    : arithmetic_type_(type), left_(std::move(left)), right_(std::move(right))
{}

bool ArithmeticExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (type() != other.type()) {
    return false;
  }
  auto &other_arith_expr = static_cast<const ArithmeticExpr &>(other);
  return arithmetic_type_ == other_arith_expr.arithmetic_type() && left_->equal(*other_arith_expr.left_) &&
         right_->equal(*other_arith_expr.right_);
}
AttrType ArithmeticExpr::value_type() const
{
  if(left_->value_type() == AttrType::NULLS || (right_ && right_->value_type() == AttrType::NULLS)) {
    return AttrType::NULLS;
  }
  if (!right_) {
    return left_->value_type();
  }

  if (left_->value_type() == AttrType::INTS && right_->value_type() == AttrType::INTS &&
      arithmetic_type_ != Type::DIV) {
    return AttrType::INTS;
  }

  return AttrType::FLOATS;
}

RC ArithmeticExpr::calc_value(const Value &left_value, const Value &right_value, Value &value) const
{
  RC rc = RC::SUCCESS;

  const AttrType target_type = value_type();
  value.set_type(target_type);

  switch (arithmetic_type_) {
    case Type::ADD: {
      Value::add(left_value, right_value, value);
    } break;

    case Type::SUB: {
      Value::subtract(left_value, right_value, value);
    } break;

    case Type::MUL: {
      Value::multiply(left_value, right_value, value);
    } break;

    case Type::DIV: {
      Value::divide(left_value, right_value, value);
    } break;

    case Type::NEGATIVE: {
      Value::negative(left_value, value);
    } break;

    default: {
      rc = RC::INTERNAL;
      LOG_WARN("unsupported arithmetic type. %d", arithmetic_type_);
    } break;
  }
  return rc;
}

template <bool LEFT_CONSTANT, bool RIGHT_CONSTANT>
RC ArithmeticExpr::execute_calc(
    const Column &left, const Column &right, Column &result, Type type, AttrType attr_type) const
{
  RC rc = RC::SUCCESS;
  switch (type) {
    case Type::ADD: {
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, AddOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, AddOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
    } break;
    case Type::SUB:
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, SubtractOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, SubtractOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    case Type::MUL:
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, MultiplyOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, MultiplyOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    case Type::DIV:
      if (attr_type == AttrType::INTS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, int, DivideOperator>(
            (int *)left.data(), (int *)right.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, DivideOperator>(
            (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    case Type::NEGATIVE:
      if (attr_type == AttrType::INTS) {
        unary_operator<LEFT_CONSTANT, int, NegateOperator>((int *)left.data(), (int *)result.data(), result.capacity());
      } else if (attr_type == AttrType::FLOATS) {
        unary_operator<LEFT_CONSTANT, float, NegateOperator>(
            (float *)left.data(), (float *)result.data(), result.capacity());
      } else {
        rc = RC::UNIMPLEMENTED;
      }
      break;
    default: rc = RC::UNIMPLEMENTED; break;
  }
  if (rc == RC::SUCCESS) {
    result.set_count(result.capacity());
  }
  return rc;
}

RC ArithmeticExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;

  Value left_value;
  Value right_value;
  
  rc = left_->get_value(tuple, left_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }
  if (!right_) {
    return calc_value(left_value, Value(0), value);
  }
  rc = right_->get_value(tuple, right_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
    return rc;
  }
  return calc_value(left_value, right_value, value);
}

RC ArithmeticExpr::get_column(Chunk &chunk, Column &column)
{
  RC rc = RC::SUCCESS;
  if (pos_ != -1) {
    column.reference(chunk.column(pos_));
    return rc;
  }
  Column left_column;
  Column right_column;

  rc = left_->get_column(chunk, left_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get column of left expression. rc=%s", strrc(rc));
    return rc;
  }
  rc = right_->get_column(chunk, right_column);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get column of right expression. rc=%s", strrc(rc));
    return rc;
  }
  return calc_column(left_column, right_column, column);
}

RC ArithmeticExpr::calc_column(const Column &left_column, const Column &right_column, Column &column) const
{
  RC rc = RC::SUCCESS;

  const AttrType target_type = value_type();
  column.init(target_type, left_column.attr_len(), max(left_column.count(), right_column.count()));
  bool left_const  = left_column.column_type() == Column::Type::CONSTANT_COLUMN;
  bool right_const = right_column.column_type() == Column::Type::CONSTANT_COLUMN;
  if (left_const && right_const) {
    column.set_column_type(Column::Type::CONSTANT_COLUMN);
    rc = execute_calc<true, true>(left_column, right_column, column, arithmetic_type_, target_type);
  } else if (left_const && !right_const) {
    column.set_column_type(Column::Type::NORMAL_COLUMN);
    rc = execute_calc<true, false>(left_column, right_column, column, arithmetic_type_, target_type);
  } else if (!left_const && right_const) {
    column.set_column_type(Column::Type::NORMAL_COLUMN);
    rc = execute_calc<false, true>(left_column, right_column, column, arithmetic_type_, target_type);
  } else {
    column.set_column_type(Column::Type::NORMAL_COLUMN);
    rc = execute_calc<false, false>(left_column, right_column, column, arithmetic_type_, target_type);
  }
  return rc;
}

RC ArithmeticExpr::try_get_value(Value &value) const
{
  RC rc = RC::SUCCESS;

  Value left_value;
  Value right_value;

  rc = left_->try_get_value(left_value);
  if (rc != RC::SUCCESS) {
    LOG_WARN("failed to get value of left expression. rc=%s", strrc(rc));
    return rc;
  }

  if (right_) {
    rc = right_->try_get_value(right_value);
    if (rc != RC::SUCCESS) {
      LOG_WARN("failed to get value of right expression. rc=%s", strrc(rc));
      return rc;
    }
  }

  return calc_value(left_value, right_value, value);
}

////////////////////////////////////////////////////////////////////////////////

UnboundAggregateExpr::UnboundAggregateExpr(const char *aggregate_name, Expression *child)
    : aggregate_name_(aggregate_name), child_(child)
    {
      std::transform(aggregate_name_.begin(), aggregate_name_.end(), aggregate_name_.begin(), ::toupper);
    }

UnboundAggregateExpr::UnboundAggregateExpr(const char *aggregate_name, unique_ptr<Expression> child)
    : aggregate_name_(aggregate_name), child_(std::move(child))
{
  std::transform(aggregate_name_.begin(), aggregate_name_.end(), aggregate_name_.begin(), ::toupper);
}

////////////////////////////////////////////////////////////////////////////////
AggregateExpr::AggregateExpr(Type type, Expression *child) : aggregate_type_(type), child_(child) {}

AggregateExpr::AggregateExpr(Type type, unique_ptr<Expression> child) : aggregate_type_(type), child_(std::move(child))
{}

RC AggregateExpr::get_column(Chunk &chunk, Column &column)
{
  RC rc = RC::SUCCESS;
  if (pos_ != -1) {
    column.reference(chunk.column(pos_));
  } else {
    rc = RC::INTERNAL;
  }
  return rc;
}

bool AggregateExpr::equal(const Expression &other) const
{
  if (this == &other) {
    return true;
  }
  if (other.type() != type()) {
    return false;
  }
  const AggregateExpr &other_aggr_expr = static_cast<const AggregateExpr &>(other);
  return aggregate_type_ == other_aggr_expr.aggregate_type() && child_->equal(*other_aggr_expr.child());
}

unique_ptr<Aggregator> AggregateExpr::create_aggregator() const
{
  unique_ptr<Aggregator> aggregator;
  switch (aggregate_type_) {
    case Type::SUM: {
      aggregator = make_unique<SumAggregator>();
      break;
    }
    case Type::COUNT: {
      aggregator = make_unique<CountAggregator>();
      break;
    }
    case Type::AVG: {
      aggregator = make_unique<AvgAggregator>();
      break;
    }
    case Type::MIN: {
      aggregator = make_unique<MinAggregator>();
      break;
    }
    case Type::MAX: {
      aggregator = make_unique<MaxAggregator>();
      break;
    }
    default: {
      ASSERT(false, "unsupported aggregate type");
      break;
    }
  }
  return aggregator;
}

RC AggregateExpr::get_value(const Tuple &tuple, Value &value) const
{
  return tuple.find_cell(TupleCellSpec(name()), value);
}

RC AggregateExpr::type_from_string(const char *type_str, AggregateExpr::Type &type)
{
  RC rc = RC::SUCCESS;
  if (0 == strcasecmp(type_str, "count")) {
    type = Type::COUNT;
  } else if (0 == strcasecmp(type_str, "sum")) {
    type = Type::SUM;
  } else if (0 == strcasecmp(type_str, "avg")) {
    type = Type::AVG;
  } else if (0 == strcasecmp(type_str, "max")) {
    type = Type::MAX;
  } else if (0 == strcasecmp(type_str, "min")) {
    type = Type::MIN;
  } else {
    rc = RC::INVALID_ARGUMENT;
  }
  return rc;
}


#include "sql/parser/parse_defs.h"
#include "sql/stmt/select_stmt.h"
#include "sql/operator/logical_operator.h"
#include "sql/operator/physical_operator.h"
#include "storage/trx/trx.h"

RC SubqueryExpr::get_value(const Tuple &tuple, Value &value) const
{
  RC rc = RC::SUCCESS;
  if(!physical_operator_) {
    LOG_WARN("Subquery expression is not initialized with a physical operator");
    return RC::INVALID_ARGUMENT;
  }

  physical_operator_->set_outer_tuple(&tuple);
  if(!is_open){
    rc = physical_operator_->open(trx_);
    if(rc!=RC::SUCCESS)
    {
      LOG_WARN("failed to open physical operator of subquery. rc=%s",strrc(rc));
      return rc;
    }
    is_open = true;
  }
  rc = physical_operator_->next();
  if(rc != RC::SUCCESS) {
    LOG_WARN("failed to get next tuple from physical operator. rc=%s", strrc(rc));
    physical_operator_->close();
    is_open = false;
    return rc;
  }

  Tuple *current_tuple = nullptr;
  current_tuple = physical_operator_->current_tuple();
  if(!current_tuple) {
    LOG_WARN("failed to get current tuple from physical operator");
    physical_operator_->close();
    is_open=false;
    return RC::INVALID_ARGUMENT;
  }

  
  rc = current_tuple->cell_at(0,value);
  if(rc != RC::SUCCESS) {
    LOG_WARN("failed to get cell at index 0 from current tuple");
    physical_operator_->close();
    is_open = false;
    return rc;
  }
  if(current_tuple->cell_num()!=1)
  {
    LOG_WARN("subquery expression must return exactly one column");
    physical_operator_->close();
    is_open = false;
    return RC::MULTI_COLUMN_SUBQUERY;
  }
  return RC::SUCCESS;
}


RC SubqueryExpr::try_get_value(Value &value) const
{
  RC rc = RC::SUCCESS;
  if(!physical_operator_) {
    LOG_WARN("Subquery expression is not initialized with a physical operator");
    return RC::INVALID_ARGUMENT;
  }

  physical_operator_->set_outer_tuple(nullptr);
  if(!is_open){
    physical_operator_->open(trx_);
    is_open = true;
  }
  rc = physical_operator_->next();
  if(rc != RC::SUCCESS) {
    LOG_WARN("failed to get next tuple from physical operator. rc=%s", strrc(rc));
    physical_operator_->close();
    is_open = false;
    return rc;
  }

  Tuple *current_tuple = nullptr;
  current_tuple = physical_operator_->current_tuple();
  if(!current_tuple) {
    LOG_WARN("failed to get current tuple from physical operator");
    physical_operator_->close();
    is_open=false;
    return RC::INVALID_ARGUMENT;
  }

  
  rc = current_tuple->cell_at(0,value);
  if(rc != RC::SUCCESS) {
    LOG_WARN("failed to get cell at index 0 from current tuple");
    physical_operator_->close();
    is_open = false;
    return rc;
  }
  if(current_tuple->cell_num()!=1)
  {
    LOG_WARN("subquery expression must return exactly one column");
    physical_operator_->close();
    is_open = false;
    return RC::MULTI_COLUMN_SUBQUERY;
  }
  return RC::SUCCESS;
}



void SubqueryExpr::close()
{
  if(physical_operator_) {
    physical_operator_->close();
    is_open = false;
  }
}

SubqueryExpr:: ~SubqueryExpr() {
  if(physical_operator_) {
    physical_operator_->close();
    physical_operator_.reset();
  }
  if(select_sql_node_) {
    delete select_sql_node_;
  }
}

AttrType SubqueryExpr::value_type() const
{
  if(select_stmt_) {
    ASSERT(select_stmt_->query_expressions().size()==1,"subquery must return exactly one column when value_type()");
    return select_stmt_->query_expressions()[0]->value_type();
  }

  return AttrType::UNDEFINED;
}

unique_ptr<Expression> SubqueryExpr::copy() const
  {
    auto ptr = make_unique<SubqueryExpr>(select_sql_node_);
    select_sql_node_ = nullptr; // Prevent double deletion
    ptr->set_name(name());
    ptr->select_stmt_ = select_stmt_;
    ptr->logical_operator_ = logical_operator_;
    ptr->physical_operator_ = physical_operator_;
    return ptr;
  }


void SubqueryExpr::set_stmt(SelectStmt* stmt)
{
   select_stmt_=std::shared_ptr<SelectStmt>(stmt);
}

void SubqueryExpr::set_logical_oper(LogicalOperator* logical_oper)
{
  logical_operator_ = std::shared_ptr<LogicalOperator>(logical_oper);
}

void SubqueryExpr::set_physical_oper(PhysicalOperator* physical_oper)
{
  physical_operator_ = std::shared_ptr<PhysicalOperator>(physical_oper);
}
